package com.itheima.zookeeper;

import java.util.Map;
import java.util.stream.Collectors;

import org.I0Itec.zkclient.ZkClient;

/**
 * @author ：yuan_he_song
 */
public class MasterResolve {
    private String server = "192.168.244.130:2181";
    private ZkClient zkClient;
    private static final String rootPath = "/tuling-master";
    private static final String servicePath = rootPath + "/service";
    private String nodePath; //  /tuling-master/service0000000001
    private volatile boolean master = false;
    private static MasterResolve resolve;

    public MasterResolve(){
        zkClient = new ZkClient(server,2000, 5000);
        buildRoot();
        createServerNode();
    }

    /** 创建服务节点--临时序号节点 */
    private void createServerNode() {
        nodePath = zkClient.createEphemeralSequential(servicePath, "slave");
        System.out.println("创建server节点:" + nodePath);
        initMaster();
        initListener();
    }

    /** 监听rootPath */
    private void initListener() {
        zkClient.subscribeChildChanges(rootPath, (parentPath, currentPath) -> {
            doElection();
        });
    }

    /** 判断是否有master, 没有则选举最小节点为master */
    private void initMaster() {
        boolean existMaster  = zkClient.getChildren(rootPath)
                .stream()
                .map(p -> rootPath + "/" + p)
                .map(p -> zkClient.readData(p))
                .anyMatch(d -> "master".equals(d));
        if (!existMaster){ // 不存在master, 执行选举
            doElection();
            System.out.println("当前当选master");
        }

    }

    /** 选举master */
    private void doElection() {
       Map<String,Object> childData  = zkClient.getChildren(rootPath)
                .stream()
                .map(p -> rootPath + "/" +p)
                .collect(Collectors.toMap(p-> p, p->zkClient.readData(p)));

       if (childData.containsValue("master")){
            return; // 存在master
       }
       childData.keySet().stream().sorted().findFirst().ifPresent(p -> {
            if (nodePath.equals(p)){
                zkClient.writeData(nodePath,"master");
                master = true;
                System.out.println("当前当选master" + nodePath);
            }
       });
    }

    /** 创建根节点 */
    private void buildRoot() {
        if (!zkClient.exists(rootPath)){
            zkClient.createPersistent(rootPath);
        }
    }

    public static MasterResolve getInstance() {
        if (resolve == null) {
            resolve= new MasterResolve();
        }
        return resolve;
    }

    /** 当前节点是否是master */
    public static boolean isMaster(){
        return getInstance().master;
    }

}
